package com.central.dataManage.common.datasourcePlugin;

import com.alibaba.druid.sql.ast.SQLStatement;
import com.central.common.constant.CommonConstant;
import com.central.common.model.ColumnInfo;
import com.central.common.model.IndexInfo;
import com.central.dataManage.common.utils.PropertyUtils;
import com.central.dataManage.model.DatabaseInfo;
import com.central.common.model.DatasourceInfo;
import com.central.common.model.TableInfo;
import io.itit.itf.okhttp.FastHttpClient;
import io.itit.itf.okhttp.util.FileUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.client.HdfsAdmin;
import org.apache.hadoop.security.UserGroupInformation;

import java.io.File;
import java.io.InputStream;
import java.net.URI;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.List;

import static com.central.dataManage.common.Constants.STORE_KEYTAB_DIR;
import static com.central.dataManage.common.Constants.STORE_KEYTAB_DIR_DEFAULT;

/**
 * @author Tindy
 * @date 2021/7/20
 * @describe
 */
public class HdfsDatasourcePlugin extends BaseDatasourcePlugin {
    private final IllegalArgumentException exception=new IllegalArgumentException("hdfs数据源不支持该操作");

    public HdfsDatasourcePlugin(DatasourceInfo datasource) {
        super(datasource);
    }

    @Override
    public String getJdbcUrl() {
        throw exception;
    }

    @Override
    public Connection getConn() throws SQLException {
        throw exception;
    }

    public FileSystem getFs() {
        Configuration conf = new Configuration();
        //文件（目录）修改权限需要配置
        conf.set("fs.permissions.umask-mode", "000");
        FileSystem fs = null;
        try {
            if (StringUtils.isNotBlank(getDatasource().getKeytabUrl()) && StringUtils.isNotBlank(getDatasource().getKrb5ConfUrl())) {
                synchronized (CommonConstant.KRB5CONF_LOCK) {
                    UserGroupInformation.setLoginUser(null);
                    System.clearProperty("java.security.krb5.conf");
                    String krb5ConfPath = getKrb5ConfPath(getDatasource());
                    System.setProperty("java.security.krb5.conf", krb5ConfPath);
                    conf.set("hadoop.security.authentication", "Kerberos");
                    UserGroupInformation.setConfiguration(conf);
                    String keytabPath = getKeyTabPath(getDatasource());
                    UserGroupInformation.loginUserFromKeytab(getDatasource().getUsername(), keytabPath);
                    fs = FileSystem.get(new URI(getDatasource().getHost()), conf);
                }
            }else {
                fs = FileSystem.get(new URI(getDatasource().getHost()), conf);
            }
        } catch (Exception exception) {
            exception.printStackTrace();
        }
        return fs;
    }

    public HdfsAdmin getDfsAdmin() {
        HdfsAdmin hdfsAdmin = null;
        if (StringUtils.isNotBlank(getDatasource().getKeytabUrl())) {
            try {
                Configuration conf = new Configuration();
                conf.set("hadoop.security.authentication", "Kerberos");
                UserGroupInformation.setConfiguration(conf);
                String keytabPath = getKeyTabPath(getDatasource());
                UserGroupInformation.loginUserFromKeytab(getDatasource().getUsername(), keytabPath);
                hdfsAdmin = new HdfsAdmin(new URI(getDatasource().getHost()), conf);
            } catch (Exception exception) {
                exception.printStackTrace();
            }
        }
        return hdfsAdmin;
    }

    private String getKeyTabPath(DatasourceInfo datasourceInfo) throws Exception {
        File file = new File(PropertyUtils.getString(STORE_KEYTAB_DIR,STORE_KEYTAB_DIR_DEFAULT) + File.separator + datasourceInfo.findKeytabName());
        if (!file.exists()) {
            InputStream is = FastHttpClient.get().url(datasourceInfo.getKeytabUrl()).build().execute().byteStream();
            FileUtil.saveContent(is, file);
            is.close();
        }
        return file.getPath();
    }

    private String getKrb5ConfPath(DatasourceInfo datasourceInfo) throws Exception {
        File file = new File(PropertyUtils.getString(STORE_KEYTAB_DIR,STORE_KEYTAB_DIR_DEFAULT) + File.separator + datasourceInfo.findKrb5ConfName());
        if (!file.exists()) {
            InputStream is = FastHttpClient.get().url(datasourceInfo.getKrb5ConfUrl()).build().execute().byteStream();
            FileUtil.saveContent(is, file);
            is.close();
        }
        return file.getPath();
    }

    @Override
    public List<DatabaseInfo> getDatabases(Connection conn) throws SQLException {
        throw exception;
    }

    @Override
    public List<TableInfo> getTables(DatabaseInfo database, Connection conn) throws SQLException {
        throw exception;
    }

    @Override
    public List<ColumnInfo> getColumns(DatabaseInfo database, TableInfo table, Connection conn) throws SQLException {
        throw exception;
    }

    @Override
    public List<IndexInfo> getIndexs(DatabaseInfo database, TableInfo table, Connection conn) throws SQLException {
        throw exception;
    }

    @Override
    public void createNewDatabase(String dbName) throws SQLException {
        throw exception;
    }

    @Override
    public void dropDatabase(String dbName) throws SQLException {
        throw exception;
    }

    @Override
    public void dropTable(String dbName, String tableName) throws SQLException {
        throw exception;
    }

    @Override
    public List<SQLStatement> checkCreateTableDdl(String ddl) throws SQLException {
        throw exception;
    }

    @Override
    public void createTableByDdl(DatabaseInfo database, List<SQLStatement> sqlStatements) throws SQLException {
        throw exception;
    }

    @Override
    public String generateDdlSql(TableInfo table) {
        throw exception;
    }

    @Override
    public String getTableNameByDdl(String ddl) {
        throw exception;
    }

    @Override
    public String getTableNameByModifyDdl(SQLStatement sqlStatement) {
        throw exception;
    }

    @Override
    public String generateModifySql(TableInfo table, TableInfo oldTable) throws Exception {
        throw exception;
    }

    @Override
    public void modifyTableByDdl(DatabaseInfo database, List<SQLStatement> sqlStatements) throws SQLException {
        throw exception;
    }

    @Override
    public List<SQLStatement> checkModifyTableDdl(String ddl, String tableName) {
        throw exception;
    }

    @Override
    public List<String> getDataTypes() {
        throw exception;
    }

    @Override
    public String format(String sql) {
        throw exception;
    }

    @Override
    public String getDatasourceDriver() {
        throw exception;
    }
}
